/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package mr.scan;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import util.IntPair;

/**
 *
 * @author tung
 * input: a set of file. 
 * Each file contains many rows. 
 * Each row contains a set of integers in which the first integer is the index of row.
 * For example: a file with 3 rows
 * <content>
 * 1 2 5 3 6 7 9
 * 2 5 3 6 2 9 1 5 6 3
 * 3 7 3 2 4
 * </content>
 */
public class ScanTung extends Configured implements Tool {

    /* The number of lines. Currently, I don't know how to automatically get this value and pass it to Mappers.
     * 
     * There is a trade-off between the number of row and the number of element in the row.
     */
    
    public static class AccReduce extends MapReduceBase implements Reducer<IntWritable, IntPair, IntWritable, IntWritable> {

        /*
         * This reducer is not good. It requires two iterations.
         * Should find an alternative algorithm which is more optimizer.
         */
        @Override
		public void reduce(IntWritable key, Iterator<IntPair> values, OutputCollector<IntWritable, IntWritable> output, Reporter reporter) throws IOException {
            int prefixsum = 0;
            int startIndex = 0;
            
            List<IntPair> lsTemp = new ArrayList<IntPair>();

            // calculate the accumulative prefixsum and startIndex for the current row.
            while (values.hasNext()) {
                IntPair itTemp = values.next();
                if (itTemp.getIndex() == 0) {
                    prefixsum += itTemp.getValue();
                } else if (itTemp.getIndex() == -1) {
                    startIndex += itTemp.getValue();
                }
                else {
                    System.out.println("(" + itTemp.getIndex() + "," + itTemp.getValue() + ")");
                    lsTemp.add(new IntPair(itTemp.getIndex(), itTemp.getValue()));                    
                }
            }

            // update values of elements in the row to get the final result.
            for (IntPair temp : lsTemp) {
                // accumulation operator: addition
                output.collect(new IntWritable(temp.getIndex() + startIndex), new IntWritable(temp.getValue() + prefixsum));
            }
        }
    }

    public static class LocalScanMap extends MapReduceBase implements Mapper<Object, Text, IntWritable, IntPair> {

        private IntWritable oindex = new IntWritable();
        private IntPair ovalue = new IntPair();

        @Override
        public void map(Object key, Text value, OutputCollector<IntWritable, IntPair> output, Reporter reporter) throws IOException {
            int lineindex = 0;
            int localindex = 0;

            // the first integer is the index of row
            StringTokenizer itr = new StringTokenizer(value.toString());
            if (itr.hasMoreTokens()) {
                lineindex = Integer.parseInt(itr.nextToken());
            }

            // calculate prefix sums
            int prefixsum = 0;
            while (itr.hasMoreTokens()) {
                // local scan operator: addition
                prefixsum += Integer.parseInt(itr.nextToken());
                localindex++;
                
                oindex.set(lineindex);

                ovalue.setIndex(localindex);
                ovalue.setValue(prefixsum);
                
                System.out.println("(" + lineindex + ", [" + localindex + "," + prefixsum + "])");
                output.collect(oindex, ovalue);                
            }
            
            /* 
             * localindex can be used to calculate the indeces of remaining elements in other rows. It is the number of element in the current row
             * send the prefixsum to other rows to calculate the final result 
             * 
             */
            for (int i = lineindex + 1; i <= numLines; i++) {
                // send the prefixsum
                ovalue.setIndex(0);
                ovalue.setValue(prefixsum);
                output.collect(new IntWritable(i), ovalue);
                System.out.println("(" + i + ", [0" + "," + prefixsum + "])");
                                
                // send the localindex
                output.collect(new IntWritable(i), new IntPair(-1, localindex));                                 
                System.out.println("("+ i + ", [-1" + "," + localindex + "])");

            }
        }
    }

    public static int numLines = 5;

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new ScanTung(), args);
        System.exit(res);
    }

    @Override
    public int run(String[] args) throws Exception {
        if (args.length != 3) {
            System.err.println("Usage: scan <in> <out> <rednum>");
            System.exit(2);
        }
        JobConf job = new JobConf(getConf(), NumScan.class);
        job.setJarByClass(NumScan.class);
        job.setMapperClass(LocalScanMap.class);
        job.setReducerClass(AccReduce.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(IntPair.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        job.setNumReduceTasks(Integer.parseInt(args[2]));
        try {
            JobClient.runJob(job);
        } finally {
        }
        return 0;
    }
}